package qinggui.mlz.console.rocketmq.listener.transactionlistener;

import com.alibaba.fastjson2.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import qinggui.mlz.module.category.mapper.CategoryMapper;
import qinggui.mlz.module.rocketmq.event.CategoryDeleteEvent;

import javax.annotation.Resource;
import java.util.List;

@Slf4j
@Component
@RocketMQTransactionListener
public class CategoryDeleteTransactionListener implements RocketMQLocalTransactionListener {
    @Resource
    private CategoryMapper categoryMapper;

    /**
     * 本地事务：再次确认类目已软删 => 提交；否则回滚
     *
     * @param message 消息，一般是消息体，checkLocalTransaction的时候才进行解析
     * @param arg     业务对象参数
     */
    @Override
    @SuppressWarnings("unchecked")
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {
        try {
            List<Long> deletesIds = (List<Long>) arg;

            if (deletesIds == null || deletesIds.isEmpty()) {
                return RocketMQLocalTransactionState.COMMIT;
            }
            // 只要还有一条类目存在（没被软删）就回滚
            boolean anyExist = deletesIds.stream()
                    .anyMatch(id -> categoryMapper.getById(id) != null);
            return anyExist ? RocketMQLocalTransactionState.ROLLBACK
                    : RocketMQLocalTransactionState.COMMIT;

        } catch (Exception e) {
            log.error("批量删除类目事务异常", e);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    /**
     * Broker 回查（15 次，默认 1 分钟一次）, 判断本地事务状态，确认类目已软删 => 提交；否则回滚
     * 半消息 发到 MQ 后，如果 Producer 宕机 或 本地事务返回 UNKNOWN，
     * Broker 会主动回查 这条消息到底该 提交 还是 回滚。返回commit和rollback，不会回查
     * 与本地事务逻辑 保持一致，确保 无论回查多少次，结果都一样（幂等）。
     *
     * @param message 消息
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        try {
            // 1. 反序列化
            CategoryDeleteEvent event = JSON.parseObject(
                    new String((byte[]) message.getPayload()),
                    CategoryDeleteEvent.class);
            List<Long> categoryIds = event.getCategoryIds();
            if (categoryIds == null || categoryIds.isEmpty()) {
                // 没有类目需要校验，直接提交
                return RocketMQLocalTransactionState.COMMIT;
            }

            // 2. 只要还有一条类目没被软删（仍能查到），就回滚
            boolean anyExist = categoryIds.stream()
                    .anyMatch(id -> categoryMapper.getById(id) != null);
            return anyExist ? RocketMQLocalTransactionState.ROLLBACK
                    : RocketMQLocalTransactionState.COMMIT;

        } catch (Exception e) {
            log.error("回查本地事务异常", e);
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
}

